package com.ylkj.process;


import com.hwl.datahub.SchemaVo;
import com.ylkj.bus.MessageContent;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.util.List;

/**
 * @Author: wengzx
 * @Date: 2025/4/21
 * @Desc:
 */
@Slf4j
public class UnpackConvertProcess extends ProcessFunction<List<SchemaVo>, MessageContent>  {

    private static final String DATA = "data";
    private static final String PROPERTIES = "properties";

    @Override
    public void processElement(List<SchemaVo> values, ProcessFunction<List<SchemaVo>, MessageContent>.Context ctx, Collector<MessageContent> out) throws Exception {
        if(values == null || values.isEmpty()){
            return;
        }
        while (!values.isEmpty()) {
            SchemaVo remove = values.remove(values.size() - 1);
            MessageContent messageContent = convert(remove);
            out.collect(messageContent);
        }
    }
    private MessageContent convert(SchemaVo schemaVo){
        String propertites = schemaVo.get(PROPERTIES);
        String data = schemaVo.get(DATA);
        if(StringUtils.isBlank(propertites) || StringUtils.isBlank(data) ){
            log.debug("原始报文数据propertites:{},data:{}为空",propertites,data);
        }
        MessageContent mqttMsg = new MessageContent(data,propertites);
        return mqttMsg;
    }
}

